package com.dahua.messagesyncbj.customer;

import com.alibaba.fastjson2.JSON;
import com.dahua.messagesyncbj.config.redis.RedisUtil;
import com.dahua.messagesyncbj.config.websocket.WebSocketServer;
import com.dahua.messagesyncbj.pojo.Body;
import com.dahua.messagesyncbj.pojo.MeasureInfo;
import com.dahua.messagesyncbj.runner.KafkaHandleRunner;
import com.dahua.messagesyncbj.runner.SyncHandleRunner;
import com.dahua.messagesyncbj.service.EquipmentService;
import com.dahua.messagesyncbj.service.FaultService;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.redisson.api.RedissonClient;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.util.List;

import static com.dahua.messagesyncbj.business.Constants.IS_INITIALIZE;

@Component
@Slf4j
public class KafkaConsumer {
    @Resource
    private RedisUtil redisUtil;
    @Resource
    private RedissonClient redissonClient;
    @Resource
    private WebSocketServer webSocketServer;

    @Resource
    private EquipmentService equipmentService;
    @Resource
    private FaultService faultService;

    @Resource
    private KafkaHandleRunner kafkaHandleRunner;
    @Resource
    private SyncHandleRunner syncHandleRunner;

    @KafkaListener(topics = {"CMODE_DATA"}, groupId = "CONSUMER_GROUP_prod")
    public void onMessage1(List<ConsumerRecord<String, String>> consumerRecords, Acknowledgment ack) {
        //redis
        if (IS_INITIALIZE) {
            if (consumerRecords.size() >= 1000) {
                KafkaHandleRunner kafkaHandleRunner = new KafkaHandleRunner(redisUtil, redissonClient,
                        equipmentService, faultService, consumerRecords);
                Thread thread = new Thread(kafkaHandleRunner);
                thread.start();
            } else {
                kafkaHandleRunner.handle(consumerRecords);
            }

            for (ConsumerRecord<String, String> item : consumerRecords) {
                MeasureInfo info = JSON.parseObject(item.value(), MeasureInfo.class);
                Body body = info.getBody();
                try {
                    webSocketServer.sendMessage(body.getMacid(), JSON.toJSONString(body));
                } catch (Exception e) {
                    log.info(e.getMessage());
                }
            }
            ack.acknowledge();
        }
    }

    /*  consume message's updated equipment status information*/
    @KafkaListener(topics = {"SYNC_DATA"}, groupId = "CONSUMER_GROUP_prod")
    public void onMessage2(List<ConsumerRecord<String, String>> consumerRecords, Acknowledgment ack) {
        try{
            syncHandleRunner.handle(consumerRecords);
        }catch (Exception e){
            log.error(e.getMessage());
        }
        ack.acknowledge();
    }
}
